package com.dap.dubbo.rpc.utils;

import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.Result;
import com.alibaba.dubbo.rpc.RpcResult;
import com.dap.dubbo.rpc.config.RedisConfig;
import com.dap.dubbo.rpc.filter.ValidationProviderFilter;
import com.google.common.util.concurrent.RateLimiter;
import groovy.lang.GroovyShell;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.ZParams;

import java.util.Collections;
import java.util.List;
import java.util.UUID;

public class ServiceRateLimiter {

    private static final Logger LOGGER = LoggerFactory.getLogger(ValidationProviderFilter.class);

    private static final String Service_Prefix="ServiceLimiter:";

    private static final String Config_Prefix="ConfigLimiter:";

    /**
     * 服务降级并发限制
     */
    private  RateLimiter rateLimiter = RateLimiter.create(500);

    private GroovyShell groovyShell = new GroovyShell();

    private int limit=1000;

    private int timeout=1000;

    private String script="return null";

    private Invoker<?> invoker;

    private Invocation invocation;


    public ServiceRateLimiter(Invoker<?> invoker, Invocation invocation) {
        this.invoker = invoker;
        this.invocation = invocation;
    }

    public Result execute() {
        System.out.println("redis===init==execute");
        /**
         * 获取资源信息
         */
        String resourceName= ServiceRateLimiter.getResourceName(invoker,invocation);

        Jedis jedis = RedisConfig.getJedis();
        if(jedis==null)return invoker.invoke(invocation);
        Transaction scan = jedis.multi();
        scan.hget(Config_Prefix+resourceName,"limit");
        scan.hget(Config_Prefix+resourceName,"timeout");
        scan.hget(Config_Prefix+resourceName,"fallback");
        List<Object> results = scan.exec();
        results.removeAll(Collections.singleton(null));
        Transaction update = jedis.multi();
        if(results.size()==0 ||results.get(0)==null || StringUtils.isEmpty(String.valueOf(results.get(0)))){
            update.hset(Config_Prefix+resourceName,"limit",String.valueOf(limit));
        }else{
            limit=Integer.valueOf(String.valueOf(results.get(0)));
        }
        if(results.size()==0 ||results.get(1)==null && StringUtils.isEmpty(String.valueOf(results.get(1)))){
            update.hset(Config_Prefix+resourceName,"timeout",String.valueOf(timeout));
        }else{
            limit=Integer.valueOf(String.valueOf(results.get(1)));
        }
        if(results.size()==0 || results.get(2)==null && StringUtils.isEmpty(String.valueOf(results.get(2)))){
            update.hset(Config_Prefix+resourceName,"fallback",script);
        }else{
            script=String.valueOf(results.get(2));
        }
        update.exec();
        String token = acquireTokenFromBucket(resourceName,jedis,limit, timeout);
        if (jedis != null) {
            jedis.close();
        }
        if(token==null){
            /**
             * 降级服务
             */
            if(tryAcquire()){
                if(StringUtils.isEmpty(script)){
                    LOGGER.info("服务降级");
                    return  new RpcResult("降级服务");
                }else {
                    Object result = groovyShell.evaluate(" return null ");
                    return  new RpcResult(result);
                }
             }else{
                /**
                 * 拒绝服务
                 */
                return new RpcResult(new Exception("拒绝服务"));
            }
        }else{
            return invoker.invoke(invocation);
        }

    }


    private static String acquireTokenFromBucket(String resource,Jedis jedis, int limit, long timeout) {
        /**
         * 标识
         */
        String identifier = UUID.randomUUID().toString();
        /**
         * 当前时间毫秒数
         */
        long now = System.currentTimeMillis();

        Transaction transaction = jedis.multi();

        String BUCKET_MONITOR=Service_Prefix+resource+"-MONITOR";

        transaction.zremrangeByScore(BUCKET_MONITOR.getBytes(), "-inf".getBytes(),
                String.valueOf(now - timeout).getBytes());

        ZParams params = new ZParams();
        params.weightsByDouble(1.0, 0.0);
        String BUCKET=Service_Prefix+resource+"-BUCKET";
        /*
          服务具体情况赋值
         */
        transaction.zinterstore(BUCKET, params, BUCKET, BUCKET_MONITOR);
       // 服务访问总量
        String BUCKET_COUNT=Service_Prefix+resource+"-COUNT";
        transaction.incr(BUCKET_COUNT);
        List<Object> results = transaction.exec();
        /**
         * 桶数
         */
        long counter = (Long) results.get(results.size() - 1);
        transaction = jedis.multi();
        /**
         *BUCKET_MONITOR 添加时间
         */
        transaction.zadd(BUCKET_MONITOR, now, identifier);
        /**
         * 添加元素
         */
        transaction.zadd(BUCKET, counter, identifier);
        /**
         * 添加排名
         */
        transaction.zrank(BUCKET, identifier);


        results = transaction.exec();

        // 获取排名，判断请求是否取得了信号量
        long rank = (Long) results.get(results.size() - 1);

        if (rank < limit) {
            return identifier;
        } else {
            // 没有获取到信号量，清理之前放入redis 中垃圾数据
            transaction = jedis.multi();
            transaction.zrem(BUCKET_MONITOR, identifier);
            transaction.zrem(BUCKET, identifier);
            transaction.exec();
            return null;
        }
    }

    private static String getResourceName(Invoker<?> invoker, Invocation invocation) {
        StringBuilder buf = new StringBuilder(64);
        buf.append(invoker.getInterface().getName())
                .append(":")
                .append(invocation.getMethodName())
                .append("(");
        boolean isFirst = true;
        for (Class<?> clazz : invocation.getParameterTypes()) {
            if (!isFirst) {
                buf.append(",");
            }
            buf.append(clazz.getName());
            isFirst = false;
        }
        buf.append(")");
        return buf.toString();
    }

    /**
     * 尝试获取令牌
     * @return
     */
    public boolean tryAcquire(){
        return rateLimiter.tryAcquire();
    }

}
